package com.shujia.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo16Student2 {

  def main(args: Array[String]): Unit = {

    /**
      *
      * 统计每科都及格的学生 [学号，姓名，班级，科目，分数]
      *
      */


    val conf: SparkConf = new SparkConf().setMaster("local[8]").setAppName("map")
    val sc: SparkContext = new SparkContext(conf)


    val students: RDD[String] = sc.textFile("spark/data/students.txt")
    val scores: RDD[String] = sc.textFile("spark/data/score.txt")
    val cource: RDD[String] = sc.textFile("spark/data/cource.txt")


    val courceKV: RDD[(String, String)] = cource.map(line => (line.split(",")(0), line))
    val scoresKV: RDD[(String, String)] = scores.map(line => (line.split(",")(1), line))


    val joinRDD: RDD[(String, (String, String))] = courceKV.join(scoresKV)


    val filterRDD: RDD[(String, (String, String))] = joinRDD.filter(kv => {
      val couId: String = kv._1

      val counInfo: String = kv._2._1


      //科目总分
      val couSumScore: Int = counInfo.split(",")(2).toInt

      val scoInfo: String = kv._2._2

      val sco: Int = scoInfo.split(",")(2).toInt


      //将没有几个的学生的科目都过滤掉
      sco >= couSumScore * 0.6
    })


    val kvRDD: RDD[(String, Int)] = filterRDD.map(kv => {
      val id: String = kv._2._2.split(",")(0)

      (id, 1)
    })


    //统计学生几个科目的数量
    val reduceRDD: RDD[(String, Int)] = kvRDD.reduceByKey(_ + _)

    //所有科目都几个的学生
    val rdd4: RDD[(String, Int)] = reduceRDD.filter(kv => kv._2 == 6)


    //取出所有科目都及格的学生的学号
    val ids: Array[String] = rdd4.map(_._1).collect()

    val scoreCouRDD: RDD[(String, (String, Int))] = joinRDD.map(kv => {
      val couName: String = kv._2._1.split(",")(1)

      val scoInfo: String = kv._2._2

      val split: Array[String] = scoInfo.split(",")

      val id: String = split(0)
      val score: Int = split(2).toInt


      (id, (couName, score))
    })

    val filterScoreRDD: RDD[(String, (String, Int))] = scoreCouRDD.filter(kv => ids.contains(kv._1))

    val stuKv: RDD[(String, String)] = students.map(line => (line.split(",")(0), line))


    //关联学生信息
    val scoStuJOinRDD: RDD[(String, ((String, Int), String))] = filterScoreRDD.join(stuKv)


    //整理数据
    scoStuJOinRDD.map(kv => {
      val id: String = kv._1
      val couName: String = kv._2._1._1
      val sco: Int = kv._2._1._2

      val stuinfo: String = kv._2._2


      s"$stuinfo,$couName,$sco"
    }).foreach(println)


  }

}
